package com.kang.consumer;

import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.List;

/**
 * 手动注入 DefaultLitePullConsumer
 */
@Component
public class SpringDefaultLitePullConsumer {

    @Value("${rocketmq.name-server}")
    private String rocketmqNameerver;

    @Value("${rocketmq.consumer.group}")
    private String consumerGroup;

    @Value("${rocketmq.consumer.topic}")
    private String topicPull;

    private DefaultLitePullConsumer litePullConsumer;

    @PostConstruct
    public void init() throws MQClientException {
        litePullConsumer = new DefaultLitePullConsumer(consumerGroup);
        litePullConsumer.setNamesrvAddr(rocketmqNameerver);
        litePullConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        litePullConsumer.subscribe(topicPull, "*");
        litePullConsumer.setAutoCommit(true); //该值默认为 true
        litePullConsumer.start();
        try {
            while (true) {
                List<MessageExt> messageExts = litePullConsumer.poll();
                doConsumeSomething(messageExts);
            }
        } finally {
            litePullConsumer.shutdown();
        }
    }
    private static void doConsumeSomething(List<MessageExt> messageExts) {
        // 真正的业务处理
        System.out.printf("消费消息条数 %s%n", messageExts.size());
        messageExts.forEach(messageExt -> {
            String string = messageExt.toString();
            System.out.println("消费消息："+string+"%n");
        });
    }

    @PreDestroy
    public void destroy() throws MQClientException {
        if (litePullConsumer != null) {
            litePullConsumer.shutdown();
        }
    }

}
